package com.zjl.Kafka.第03章_操作kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 在消息发送的过程中，涉及到了 两个线程—— main 线程和 Sender 线程 。在 main 线程
 * 中创建了 一个双端队列 RecordAccumulator 。 main 线程将消息发送给 RecordAccumulator ，
 * Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker 。
 *
 * 回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元
 * 数据信息（ RecordMetadata ）和异常信息（ Exception ），如果 Exception 为 null ，说明消息发
 * 送成功，如果 Exception 不为 null ，说明消息发送失败。
 *
 * **********************************************************************************
 * 应答处理级别  ACKS = 0
 *        只要发送了，就不管了,不管回复
 * 应答处理级别  ACKS = -1  (all)  默认
 *      发送后，还要等待所有的主+副 都回应后才会认同
 * 应答处理级别  ACKS = 1
 *      发送后， 主 回应后才会认同
 * **********************************************************************************
 * 重试机制，当发送后，没有回应（应答），
 *      就会 retry 重试，但有可能已经保存了导致重复，重试可能导致数据重复和乱序
 * **********************************************************************************
 * 开启幂等操作 要求   他只能保证一个分区有序(每分区是不同的文件数据)    保存时，会发送  生产者id(每个会话随机生成的) + 顺序号
 *      1. ACKS = -1
 *      2. 开启 重试机制
 *      3. 在途请求缓冲区数量 <=5  (默认5)
 *              原因：再重试时，会将数据重新发送，会与本分区的数据进行比较，重复了，就会不做操作
 *              本分区的数据 只会保存5个，因为 5 经过大量测试，5是最好的，写死，不可修改
 * 幂等操作能解决，重试机制导致的重复和乱序，
 *      但是不能保证 服务挂掉，重启导致的重复(生产者id + 顺序号) 生产者id不一致了
 *         也可使用事务的方式保证重启不导致重复(回滚)
 * ***********************************************************************************
 * 事务 ,会在 数据中存入   __transaction_state-XX   的文件  不可操作还有类似文件（__consumer_offsets-xx   消费者偏移量文件）
 *              有50个  0-49
 *      事务id = 事务的字符串.code %50
 *  创建者执行事务流程
 *    1.  初始化事务
 *    2.  开始事务
 *    3.  将数据发送到kafka
 *    4.  提交事务   或    有异常可以中断事务
 *
 *  kafka 事务流程
 *      1.  生产者  -> kafka节点     查找事务管理器所在的节点
 *      2.  kafka主节点 -> 生产者    事务id =   事务值.hashCode % 50
 *      3.  生产者 -> 事务协调器      初始化生产者id ( __transaction_state-XX 这个文件的产生)
 *      4.  生产者 -> 事务协调器      将数据的分区信息传送到事务管理器
 *      5.  生产者  -> kafka节点     生产数据
 *      6.  kafka节点 -> 事务协调器   提交保存成功的数据分区信息
 *      7.  kafka节点 -> 生产者      返回单次数据是否成功
 *      8.  生产者 -> 事务协调器      结束事务
 *      9.  事务协调器 ->  事务状态   事务改成预提交状态
 *      10. 事务协调器-> kafka节点    给 Kafka 节点 说事务提交了
 *      11. 事务协调器 -> 事务状态    事务改成成功状态
 * ******************************************************************************************
 * 数据传输 语义
 * at most once
 *          最多一次:不管是否能接收到，数据最多只传一次。 这样数据可能会丢失,
 *          Socket, ACK=0
 *
 * at least once
 *          最少一次:消息不会丢失，如果接收不到，那么就继续发，所以会发送多次，直到收到为止，有可能出现数据重复
 *          ACK=1
 * Exactly once
 *          精准一次:消息只会一次，不会丢，也不会重复。
 *          幂等+事务+ACK=-1
 *
 *
 */
public class C_发送原理 {
    @Test
    public  void 生产者() throws ExecutionException, InterruptedException {
        Map<String,Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"82.157.71.243:9092");//连接地址
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//key 的 序列化
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//value 的 序列化
//        config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, B1_拦截器.class.getName());//添加拦截器
//        config.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, B2_自定义分区器.class.getName());//添加分区器
        System.out.println();
        //应答处理级别     0 : 只要发送了，就不管了,不管回复     -1:等待所有的主+副 都回应后才会认同   1:主 回应后才会认同
        config.put(ProducerConfig.ACKS_CONFIG, "-1");
        config.put(ProducerConfig.RETRIES_CONFIG, 5);//重试机制 重试次数
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, 5);//每一批大小  单位kb  用于存储数据时对每批的大小限制
        config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);//请求超时 时间  默认 30000毫秒（30秒）
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);//是否  true 开启幂等操作
        config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);//在途请求缓冲区  默认5
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "事务id6");//事务id

        //创建生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<>(config);
        kafkaProducer.initTransactions();//初始化事务
        try{
            kafkaProducer.beginTransaction();//开始事务
            for (int i = 0; i < 10; i++) {
                //创建数据
                ProducerRecord<String,String> record = new ProducerRecord<>(
                        "java0016"//主题名称  发送不成功改动主题名，就行了
                        ,"key" + i//数据的key
                        ,"value" + i//数据的value
                );
                //将数据发送到kafka   异步发送，其他线程发送
                Future<RecordMetadata> send = kafkaProducer.send(record, new Callback() {//回调方法
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        System.out.println("发送成功！！主题:" + metadata.topic() + "\t分区：" + metadata.partition() + "\t偏移量：" + metadata.offset());
                    }
                });

                RecordMetadata recordMetadata = send.get();//获取数据，只有获取数据后，才会继续执行

                System.out.println(recordMetadata);

                System.out.println(record);
            }
            kafkaProducer.commitTransaction();//提交事务
            System.out.println(kafkaProducer);

        }catch (Exception e){
            e.printStackTrace();
            kafkaProducer.abortTransaction();//终止事务
        }finally {
            try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); }
            //关闭对象
            kafkaProducer.close();
        }


    }
}
